package kafkacli

import (
	"context"
	"fmt"
	"strings"

	"os"

	"os/signal"
	"sync"
	"syscall"

	"log"

	"github.com/Shopify/sarama"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/util/grand"
)

func kafkaGroupConsume(topic string, consumer *Consumer, ext ...string) {

	log.Println("Starting a new Sarama consumer")

	sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)

	version, err := sarama.ParseKafkaVersion("2.1.1")
	if err != nil {
		log.Panicf("Error parsing Kafka version: %v", err)
	}

	/**
	 * Construct a new Sarama configuration.
	 * The Kafka cluster version has to be defined before the consumer/producer is initialized.
	 */
	config := sarama.NewConfig()
	config.Version = version
	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky

	config.Consumer.Offsets.Initial = sarama.OffsetOldest

	/**
	 * Setup a new Sarama consumer group
	 */
	groupName := "demo-grouup"
	if len(ext) >= 1 {
		groupName = ext[0]                                    // 指定 groupName
		config.Consumer.Offsets.Initial = sarama.OffsetNewest // 这种情况下不管历史消息
	}

	brokers := g.Config().GetString("KafkaBrokers")
	fmt.Println("KafkaBrokers ", brokers)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), groupName, config)
	if err != nil {
		log.Panicf("Error creating consumer group client: %v", err)
	}
	topics := getAllPrefix(topic)

	admin, err := sarama.NewClusterAdmin([]string{brokers}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	for _, topic := range topics {
		admin.CreatePartitions(topic, 20, nil, false)
	}

	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			// `Consume` should be called inside an infinite loop, when a
			// server-side rebalance happens, the consumer session will need to be
			// recreated to get the new claims
			if err := client.Consume(ctx, topics, consumer); err != nil {
				log.Panicf("Error from consumer: %v", err)
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				return
			}
			consumer.ready = make(chan bool)
		}
	}()

	<-consumer.ready // Await till the consumer has been set up
	log.Println("Sarama consumer up and running!...")

	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-ctx.Done():
		log.Println("terminating: context cancelled")
	case <-sigterm:
		log.Println("terminating: via signal")
	}

	wg.Wait()
	if err = client.Close(); err != nil {
		log.Panicf("Error closing client: %v", err)
	}
}

// ReceiveMessageFunc 消息的回掉函数
// 注意响应的函数，禁止再去开另外的协成单独处理，除非你很清除你在干什么。
type ReceiveMessageFunc func(sarama.ConsumerGroupSession, *sarama.ConsumerMessage) error

// Consumer represents a Sarama consumer group consumer
type Consumer struct {
	ready           chan bool
	messageCallback ReceiveMessageFunc
}

// NewConsumer NewConsumer
func NewConsumer(fn ReceiveMessageFunc) *Consumer {
	return &Consumer{
		ready:           make(chan bool),
		messageCallback: fn,
	}
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// Mark the consumer as ready
	close(consumer.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
	for message := range claim.Messages() {

		// 注意响应的函数，禁止再去开另外的协成单独处理，除非你很清除你在干什么。
		if err := consumer.messageCallback(session, message); err != nil {
			return err
		} else {

			session.MarkMessage(message, "")
		}
	}

	return nil
}

func ConsumeGetSubDomain(consumer *Consumer) {

	kafkaGroupConsume(TopicGetSubDomain, consumer)

}

func ConsumeAllLink(consumer *Consumer) {
	kafkaGroupConsume(TopicAllLink, consumer)

}

func ConsumeDcgo(consumer *Consumer) {

	kafkaGroupConsume(TopicDcgo, consumer)

}

func ConsumeValidUrlCheck(consumer *Consumer) {

	kafkaGroupConsume(TopicValidUrlCheck, consumer)

}
func ConsumeNewKeyword(consumer *Consumer) {
	kafkaGroupConsume(TopicNewKeyword, consumer, grand.Letters(4)) // 保证每个都是不同的分组，这种情况下不管历史消息
	select {}

}
